

# 基本组件： 消息中间件，任务执行单元，任务结果存储，-----接受异步任务和定时任务
# 消息中间件（Broker）：任务消息的临时存储与路由，解耦生产者和消费者----rabbitMQ 或 redis
# 工作进程（Worker）：任务的实际执行单元，支持分布式部署。
# 结果后端（Backend）：存储任务状态（PENDING/SUCCESS/FAILURE）和结果。
# 调度器（Beat）：触发定时任务（周期性/Crontab）



# 异步任务基于生产者消费者模型实现的

# 生产者： 链接 MQ 向 MQ 插入数据 
# 消费者： 链接 MQ 监听队列，内部实现 callback ， 如果想实现异步，还需要在 callback 中用 并发的语法 实现异步请求






import celery 
from celery.result import AsyncResult
import time
backend = "redis://"
broker = "redis://"

app = celery.Celery("", backend=backend, broker=broker, include=[
    "任务文件名.任务方法名", # 多个
])

# 任务定义与发布
@app.task  
def add(x, y):  
    return x + y  
result = add.delay(4, 6)  #  delay 发布任务至Broker，返回AsyncResult:cite[4]    函数名.delay(参数)


# 任务调度与执行
#     异步任务：Worker 从 Broker 拉取任务 → Pool 分配子进程执行 → 结果存入 Backend2。
#     定时任务：Beat 生成任务消息 → 推送至 Broker → Worker 消费执行


# 结果查询
if result.ready():  # 检查任务是否完成  
    print(result.get())  # 获取结果:cite[5]  


# celery worker -A celery_app_task -l info  ;  -A 任务名称(py文件名称)， 日志级别    
# 链接消息中间件， 创建队列， 监听队列，创建多个 worker 监听任务



from celery_tasks.celery import cel  # cel  --> app 就是 celery 的实例

async_result = AsyncResult(id="562834c6-e4be-46d2-908a-b102adbbf390", app=cel)
if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除,执行完成，结果不会自动删除
    # async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
    # async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。
elif async_result.failed():
    print('执行失败')
elif async_result.status == 'PENDING':
    print('任务等待中被执行')
elif async_result.status == 'RETRY':
    print('任务异常后正在重试')
elif async_result.status == 'STARTED':
    print('任务已经开始被执行')


# celery worker -A celery_task -l info -P eventlet  使用 eventlet 开携程    












# ------------------------ 定时任务 -------------------------
from datetime import datetime

v1 = datetime(2020, 3, 11, 16, 19, 00)   # 本地时间
print(v1) 
v2 = datetime.utcfromtimestamp(v1.timestamp())  # 国标时间
print(v2)

result = send_email.apply_async(args=["egon",], eta=v2)    # apply_async 为定时任务  本质是 delay 但是它接收了时间参数
print(result.id)



ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay
result = send_email.apply_async(args=["egon"], eta=task_time)








from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[
    'celery_tasks.task01',
    'celery_tasks.task02',
])
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False
cel.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_tasks.task01.send_email',
        # 每隔2秒执行一次
        # 'schedule': 1.0,    1秒
        # 'schedule': crontab(minute="*/1"),  每一分钟， crontab 是个函数
        'schedule': timedelta(seconds=6),   #  6秒
        # 传递参数
        'args': ('张三',)
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_tasks.task01.send_email',
    #     每年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': ('张三',)
    # },
} 
